import org.apache.spark.{SparkConf, SparkContext}

object TopSalaryEmployees {
  def main(args: Array[String]): Unit = {
    // 初始化Spark配置
    val conf = new SparkConf()
      .setAppName("TopSalaryEmployees")
      .setMaster("local[*]")  // 本地模式，使用所有可用核心

    val sc = new SparkContext(conf)

    try {
      // 1. 读取CSV文件（注意Windows路径需转义或使用原始字符串）
      val filePath = raw"C:\Employee_salary_first_half.csv"  // 使用raw避免转义
      val first_half = sc.textFile(filePath)

      // 2. 去除首行标题
      val drop_first = first_half.mapPartitionsWithIndex { (idx, iter) =>
        if (idx == 0) iter.drop(1)  // 如果是第一个分区（首行），跳过
        else iter
      }

      // 3. 解析数据：提取姓名（第2列）和薪资（第7列），并转换类型
      val split_first = drop_first.map { line =>
        val fields = line.split(",")
        // 确保字段足够且可转换，添加错误处理
        try {
          val name = fields(1)          // 第2列：员工姓名
          val salary = fields(6).toLong  // 第7列：实际薪资（转为Long）
          (name, salary)
        } catch {
          case e: Exception =>
            println(s"数据解析错误: $line")
            ("Error", 0L)  // 返回默认值，后续可过滤
        }
      }.filter(_._2 > 0)  // 过滤无效数据

      // 4. 按薪资降序排序，取前3名
      val topEmployees = split_first
        .sortBy(_._2, ascending = false)  // 按薪资降序
        .take(3)                          // 取前3条

      // 5. 打印结果
      println("上半年实际薪资排名前3的员工：")
      topEmployees.foreach { case (name, salary) =>
        println(s"姓名: $name, 薪资: $salary")
      }

    } finally {
      // 确保关闭SparkContext
      sc.stop()
    }
  }
}
